package com.mini.grpc.client.netty;

import com.mini.grpc.client.CallOptions;
import com.mini.grpc.client.ClientCall;
import com.mini.grpc.common.Metadata;
import com.mini.grpc.common.MethodDescriptor;
import com.mini.grpc.common.Status;
import com.mini.grpc.common.StatusRuntimeException;
import com.mini.grpc.transport.ClientStream;
import com.mini.grpc.transport.ClientStreamListener;
import com.mini.grpc.transport.ClientTransport;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.Executor;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;

/**
 * NettyClientCall是基于Netty的客户端调用实现。
 *
 * @param <ReqT> 请求类型
 * @param <RespT> 响应类型
 */
public class NettyClientCall<ReqT, RespT> implements ClientCall<ReqT, RespT> {
    
    private static final Logger logger = LoggerFactory.getLogger(NettyClientCall.class);
    
    private final MethodDescriptor<ReqT, RespT> method;
    private final CallOptions callOptions;
    private final ClientTransport transport;
    private final AtomicReference<ClientStream<ReqT, RespT>> stream = new AtomicReference<>();
    private final AtomicBoolean cancelled = new AtomicBoolean(false);
    private final AtomicBoolean started = new AtomicBoolean(false);
    private final AtomicBoolean halfClosed = new AtomicBoolean(false);
    private Listener<RespT> responseListener;
    
    /**
     * 创建一个新的NettyClientCall
     *
     * @param method 方法描述符
     * @param callOptions 调用选项
     * @param transport 客户端传输
     */
    public NettyClientCall(MethodDescriptor<ReqT, RespT> method, CallOptions callOptions, ClientTransport transport) {
        System.out.println("=== NettyClientCall constructor called ===");
        System.out.println("Method: " + method.getFullMethodName());
        System.out.println("Transport: " + transport.getClass().getName());
        
        this.method = method;
        this.callOptions = callOptions;
        this.transport = transport;
        
        System.out.println("NettyClientCall created: " + this);
    }
    
    @Override
    public void start(Listener<RespT> responseListener, Metadata headers) {
        System.out.println("=== NettyClientCall.start ENTRY ===");
        System.out.println("Method: " + method.getFullMethodName());
        
        logger.info("=== NettyClientCall.start called for method: {} ===", method.getFullMethodName());
        
        try {
            System.out.println("Checking started flag...");
            if (!started.compareAndSet(false, true)) {
                System.out.println("Call already started!");
                logger.error("Call already started for method: {}", method.getFullMethodName());
                throw new IllegalStateException("Call already started");
            }
            
            System.out.println("Setting response listener...");
            this.responseListener = responseListener;
            
            logger.info("Checking deadline for method: {}", method.getFullMethodName());
            
            // 检查截止时间
            Long deadlineNanoTime = callOptions.getDeadlineNanoTime();
            if (deadlineNanoTime != null && deadlineNanoTime < System.nanoTime()) {
                System.out.println("Deadline exceeded!");
                logger.warn("Deadline exceeded for method: {}", method.getFullMethodName());
                cancel("Deadline exceeded", null);
                responseListener.onClose(Status.DEADLINE_EXCEEDED.withDescription("Deadline exceeded"), new Metadata());
                return;
            }
            
            System.out.println("About to create client stream...");
            logger.info("About to create client stream for method: {}", method.getFullMethodName());
            
            // 创建客户端流
            ClientStream<ReqT, RespT> clientStream = transport.newStream(
                    method,
                    headers,
                    new StreamListener(callOptions.getExecutor()));
            
            System.out.println("Client stream created!");
            logger.info("Client stream created for method: {}", method.getFullMethodName());
            
            if (!stream.compareAndSet(null, clientStream)) {
                System.out.println("Stream already created!");
                logger.error("Stream already created for method: {}", method.getFullMethodName());
                throw new IllegalStateException("Stream already created");
            }
            
            System.out.println("NettyClientCall.start completed successfully!");
            logger.info("NettyClientCall.start completed for method: {}", method.getFullMethodName());
        } catch (Exception e) {
            System.out.println("Exception in NettyClientCall.start: " + e.getMessage());
            e.printStackTrace();
            logger.error("Exception in NettyClientCall.start for method: {}", method.getFullMethodName(), e);
            if (responseListener != null) {
                responseListener.onClose(Status.INTERNAL.withCause(e), new Metadata());
            }
            throw e;
        }
    }
    
    @Override
    public void request(int numMessages) {
        ClientStream<ReqT, RespT> clientStream = stream.get();
        if (clientStream != null) {
            clientStream.request(numMessages);
        }
    }
    
    @Override
    public void sendMessage(ReqT message) {
        if (cancelled.get()) {
            throw new IllegalStateException("Call was cancelled");
        }
        
        if (!started.get()) {
            throw new IllegalStateException("Call not started");
        }
        
        if (halfClosed.get()) {
            throw new IllegalStateException("Call was half-closed");
        }
        
        ClientStream<ReqT, RespT> clientStream = stream.get();
        if (clientStream != null) {
            clientStream.sendMessage(message);
        }
    }
    
    @Override
    public void halfClose() {
        if (cancelled.get()) {
            return;
        }
        
        if (!started.get()) {
            throw new IllegalStateException("Call not started");
        }
        
        if (halfClosed.compareAndSet(false, true)) {
            ClientStream<ReqT, RespT> clientStream = stream.get();
            if (clientStream != null) {
                clientStream.halfClose();
            }
        }
    }
    
    @Override
    public void cancel(String message, Throwable cause) {
        if (cancelled.compareAndSet(false, true)) {
            Status status = Status.CANCELLED;
            if (message != null) {
                status = status.withDescription(message);
            }
            if (cause != null) {
                status = status.withCause(cause);
            }
            
            ClientStream<ReqT, RespT> clientStream = stream.getAndSet(null);
            if (clientStream != null) {
                clientStream.cancel(status, new Metadata());
            }
            
            if (responseListener != null) {
                responseListener.onClose(status, new Metadata());
            }
        }
    }
    
    @Override
    public boolean isCancelled() {
        return cancelled.get();
    }
    
    /**
     * 客户端流监听器，用于接收来自服务端的响应和事件
     */
    private class StreamListener implements ClientStreamListener<RespT> {
        
        private final Executor executor;
        
        /**
         * 创建一个新的StreamListener
         *
         * @param executor 执行器
         */
        public StreamListener(Executor executor) {
            this.executor = executor != null ? executor : DirectExecutor.INSTANCE;
        }
        
        @Override
        public void headersReceived(Metadata headers) {
            if (cancelled.get()) {
                return;
            }
            
            try {
                executor.execute(() -> {
                    if (cancelled.get()) {
                        return;
                    }
                    
                    try {
                        responseListener.onHeaders(headers);
                    } catch (Throwable t) {
                        logger.error("Failed to process headers", t);
                        cancel("Failed to process headers", t);
                    }
                });
            } catch (Throwable t) {
                logger.error("Failed to execute headers callback", t);
                cancel("Failed to execute headers callback", t);
            }
        }
        
        @Override
        public void messageReceived(RespT message) {
            if (cancelled.get()) {
                return;
            }
            
            try {
                executor.execute(() -> {
                    if (cancelled.get()) {
                        return;
                    }
                    
                    try {
                        responseListener.onMessage(message);
                    } catch (Throwable t) {
                        logger.error("Failed to process message", t);
                        cancel("Failed to process message", t);
                    }
                });
            } catch (Throwable t) {
                logger.error("Failed to execute message callback", t);
                cancel("Failed to execute message callback", t);
            }
        }
        
        @Override
        public void closed(Status status, Metadata trailers) {
            if (cancelled.get()) {
                return;
            }
            
            try {
                executor.execute(() -> {
                    if (cancelled.get()) {
                        return;
                    }
                    
                    cancelled.set(true);
                    stream.set(null);
                    
                    try {
                        responseListener.onClose(status, trailers);
                    } catch (Throwable t) {
                        logger.error("Failed to process close", t);
                    }
                });
            } catch (Throwable t) {
                logger.error("Failed to execute close callback", t);
            }
        }
    }
    
    /**
     * 直接执行器，在当前线程中执行任务
     */
    private enum DirectExecutor implements Executor {
        INSTANCE;
        
        @Override
        public void execute(Runnable command) {
            command.run();
        }
    }
} 